package com.hung.triple.transport;

import com.google.common.collect.Maps;
import com.hung.triple.bootstrap.TripleConsumerConfig;
import com.hung.triple.common.enums.TripleRegisterType;
import com.hung.triple.loadbalancer.AbstractLoadBalancer;
import com.hung.triple.transport.netty4.NettyTransporter;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;

/**
 * 传输层公共连接池，维护 host(ip:port) -> data channel 的一对一映射关系
 *
 * <p>
 * TODO 有可能会造成内存泄露 ??????  感觉可以再加个Map映射 服务->address->Transporter
 *
 * @author Hung
 * @since 2022/9/18
 */
@Slf4j
public class TransporterPool {

    /**
     * ServiceName->(address(ip:port) -> Transporter)
     * 相同的IP可以做复用Transporter
     * <p>
     * TRANSPORTER_POOL 键位ip:port 总库，所有连接都有
     * SERVICE_TRANSPORTER_POOL 为分属不同服务的库，利用ip:port来获得Transporter
     */
    private static final Map<String, Transporter> TRANSPORTER_POOL = Maps.newConcurrentMap();
    private static final Map<String, Map<String, Transporter>> SERVICE_TRANSPORTER_POOL = Maps.newConcurrentMap();

    private static final AbstractLoadBalancer<Transporter> LOAD_BALANCER;

    static {
        try {
            LOAD_BALANCER = TripleConsumerConfig.getLoadBalancer().newInstance();
        } catch (InstantiationException | IllegalAccessException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * init connection 时调用
     * 添加传输器，由 Netty channelActive 回调方法自动添加
     */
    public static void addTransporter(String serviceName, String remoteAddress, Channel channel) {
        //没有服务就新建服务
        if (!SERVICE_TRANSPORTER_POOL.containsKey(serviceName)) {
            SERVICE_TRANSPORTER_POOL.put(serviceName, new HashMap<>());
        }

        Map<String, Transporter> transporterMap = SERVICE_TRANSPORTER_POOL.get(serviceName);
        //没有连接就创建新连接
        if (!TRANSPORTER_POOL.containsKey(remoteAddress)) {
            NettyTransporter nettyTransporter = new NettyTransporter(channel);
            transporterMap.put(remoteAddress, nettyTransporter);
            TRANSPORTER_POOL.put(remoteAddress, nettyTransporter);
            return;
        }

        //有了连接就直接注入
        transporterMap.put(remoteAddress, TRANSPORTER_POOL.get(remoteAddress));

        log.info("[Triple] remote client(address={}) connected", remoteAddress);
    }

    /**
     * subscribe 发生实例变更时调用
     */
    public static void addTransporter(String serviceName, String remoteAddress, Transporter transporter) {
        //没有服务就新建服务
        if (!SERVICE_TRANSPORTER_POOL.containsKey(serviceName)) {
            SERVICE_TRANSPORTER_POOL.put(serviceName, new HashMap<>());
        }

        Map<String, Transporter> transporterMap = SERVICE_TRANSPORTER_POOL.get(serviceName);
        transporterMap.put(remoteAddress, transporter);
        TRANSPORTER_POOL.put(remoteAddress, transporter);
    }

    public static void removeTransporter(String serviceName, String remoteAddress) {
        Map<String, Transporter> transporterMap = SERVICE_TRANSPORTER_POOL.get(serviceName);
        Transporter transporter = transporterMap.get(remoteAddress);

        if (transporter == null) {
            return;
        }
        transporterMap.remove(remoteAddress);
        //TODO 目前没有考虑到 TRANSPORTER_POOL 中的连接如果不在 SERVICE_TRANSPORTER_POOL了怎么删除，这可能会导致内存泄露
        if (!transporter.available()) {
            log.warn("[Triple] remote client(address={}) disconnected because of channel inactive", remoteAddress);
        } else {
            log.info("[Triple] remove remote client(address={})", remoteAddress);
        }
    }

    public static Transporter getTransporter(String address) {
        return TRANSPORTER_POOL.get(address);
    }

    public static Transporter getTransporter(TripleRegisterType registerType, String serviceName) {
        if (registerType == TripleRegisterType.DIRECT) {
            //TODO 待验证
//            for (String ads : registryConfig.getAddressList()) {
//                Transporter transporter = TransporterPool.getTransporter(ads);
//                if (transporter != null) {
//                    return transporter;
//                }
//            }

            return null;
        } else if (registerType == TripleRegisterType.NACOS) {
            return LOAD_BALANCER.select(new ArrayList<>(SERVICE_TRANSPORTER_POOL.get(serviceName).values()));
        } else {
            throw new RuntimeException("注册类型错误，无法获取连接");
        }
    }

    public static void destroy() {
        TRANSPORTER_POOL.forEach((s, transporter) -> {
            transporter.getChannel().close();
        });
    }


    /**
     * 指定远程地址是否存在连接器（适用于指定 IP 调用的情况）
     */
    public static boolean hasAvailableTransporter(String address) {
        Transporter transporter = TRANSPORTER_POOL.get(address);
        if (transporter == null) {
            return false;
        }
        return transporter.available();
    }

    /**
     * 当前是否存在可用连接器（正常情况下，只要有一个长连接可用就行）
     * 注：目前没有考虑Provider提供服务不相同的情况（如 A 提供服务1和2，B 提供服务3和4）
     */
    public static boolean hasAvailableTransporter() {
        //TODO 目前没有考虑Provider提供服务不相同的情况（如 A 提供服务1和2，B 提供服务3和4）
        for (Map.Entry<String, Transporter> entry : TRANSPORTER_POOL.entrySet()) {
            if (entry.getValue().available()) {
                return true;
            }
        }
        return false;
    }
}
